package com.sunday.common.reactor.e11_Multicasting_Flux_Several_Subscribers;

import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.ConnectableFlux;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * A.9. 将 Flux 多播到多个订阅者
 */
@Slf4j
public class MulticastingFluxSeveralSubscribers {

    /**
     * A.9. 将 Flux 多播到多个订阅者
     * 我想将多个订阅者连接到一个 Flux：
     * 并决定何时使用 connect（） 触发源： publish（） （返回 ConnectableFlux )
     * 并立即触发源（后期订阅者查看后期数据）：（Flux|Mono#share())
     * 并在注册了足够多的订阅者后永久连接源：.publish（）.autoConnect（n）
     * 并在订阅者高于/低于阈值时自动连接和取消源：.publish（）.refCount（n）
     * ...​但在取消之前给新订阅者一个机会：.publish（）.refCount（n， Duration）
     * 我想缓存发布服务器中的数据，并将其重播给以后的订阅者：
     * 最多元素：cache（int）n
     * 缓存在持续时间（生存时间）内看到的最新元素：（Flux|Mono#cache(Duration))
     * ...​但保留不超过元素：cache（int， Duration）n
     * 但不会立即触发源：Flux#replay（返回 ConnectableFlux)
     */

    /**
     * 并决定何时使用 connect（） 触发源： publish（） （返回 ConnectableFlux )
     */
    @Test
    public void publish() throws InterruptedException {
        log.info("并决定何时使用 connect（） 触发源： publish（） （返回 ConnectableFlux )");
        // 并决定何时使用 connect（） 触发源： publish（） （返回 ConnectableFlux )
        Flux<Integer> flux = Flux.range(1, 100).delayElements(Duration.ofMillis(100));
        // publish() 返回 ConnectableFlux
        // ConnectableFlux 是一个特殊的 Flux，它允许您将订阅者连接到它，而不是订阅源 Flux。
        // 这使您可以控制何时开始发出元素。
        // 例如，您可以在所有订阅者都已注册后开始发出元素。
        // 为此，您需要调用 ConnectableFlux#connect()。
        // 请注意，如果您在订阅者注册之前调用 connect()，则可能会丢失一些元素。
        ConnectableFlux<Integer> connectableFlux = flux.publish();
        connectableFlux.subscribe(i -> log.info("subscribe1: {}", i));
        connectableFlux.subscribe(i -> log.info("subscribe2: {}", i));
        log.info("connectableFlux.connect()");
        connectableFlux.connect();
        TimeUnit.SECONDS.sleep(1);
    }

    /**
     * 并立即触发源（后期订阅者查看后期数据）：（Flux|Mono#share())
     */
    @Test
    public void share() throws InterruptedException {
        log.info("并立即触发源（后期订阅者查看后期数据）：（Flux|Mono#share())");
        // 并立即触发源（后期订阅者查看后期数据）：（Flux|Mono#share())
        Flux<Integer> flux = Flux.range(1, 100).delayElements(Duration.ofMillis(100));
        // share() 返回 ConnectableFlux
        // share() 是 publish().refCount() 的简写形式。
        // 它将源 Flux 转换为 ConnectableFlux 并立即调用 connect()。
        // 它还将 refCount() 的订阅者数量设置为 1。
        // 这意味着只要有一个订阅者，就会保持连接。
        // 如果没有订阅者，它将取消连接。
        // 请注意，如果您在订阅者注册之前调用 connect()，则可能会丢失一些元素。
        Flux<Integer> share = flux.share();
        share.subscribe(i -> log.info("subscribe1: {}", i));
        share.subscribe(i -> log.info("subscribe2: {}", i));
        TimeUnit.SECONDS.sleep(1);
    }

    /**
     * 在 Reactor 中，flux.publish() 和 flux.share() 都用于将一个 Flux 转换为一个可供多个订阅者共享的 ConnectableFlux，但它们之间有一些区别。
     * flux.publish() 会将原始的 Flux 转换为 ConnectableFlux，但并不会自动开始发布事件，直到调用 connect() 方法。这意味着你可以在调用 connect() 方法之前对 ConnectableFlux 进行一些预处理。
     * flux.share() 是 publish().refCount() 的简写形式，它会立即开始发布事件，并且会在至少有一个订阅者时保持连接，在所有订阅者取消订阅后断开连接。
     * 因此，主要区别在于 publish() 需要手动调用 connect() 来触发事件的发布，而 share() 则是立即开始发布事件，并且会自动管理连接状态。
     */

    /**
     * 并在注册了足够多的订阅者后永久连接源：.publish（）.autoConnect（n）
     */
    @Test
    public void autoConnect() throws InterruptedException {
        log.info("并在注册了足够多的订阅者后永久连接源：.publish（）.autoConnect（n）");
        // 并在注册了足够多的订阅者后永久连接源：.publish（）.autoConnect（n）
        Flux<Integer> flux = Flux.range(1, 100).delayElements(Duration.ofMillis(100));
        // autoConnect() 返回 Flux
        // autoConnect() 是 publish().autoConnect() 的简写形式。
        // 它将源 Flux 转换为 ConnectableFlux 并立即调用 connect()。
        // 它还将 refCount() 的订阅者数量设置为 n。
        // 这意味着只要有 n 个订阅者，就会保持连接。
        // 如果没有订阅者，它将取消连接。
        // 请注意，如果您在订阅者注册之前调用 connect()，则可能会丢失一些元素。
        Flux<Integer> autoConnect = flux.publish().autoConnect(2);
        autoConnect.subscribe(i -> log.info("subscribe1: {}", i));
        autoConnect.subscribe(i -> log.info("subscribe2: {}", i));
        TimeUnit.SECONDS.sleep(1);
    }

    /**
     * 并在订阅者高于/低于阈值时自动连接和取消源：.publish（）.refCount（n）
     */
    @Test
    public void refCount() throws InterruptedException {
        log.info("并在订阅者高于/低于阈值时自动连接和取消源：.publish（）.refCount（n）");
        // 并在订阅者高于/低于阈值时自动连接和取消源：.publish（）.refCount（n）
        Flux<Integer> flux = Flux.range(1, 100).delayElements(Duration.ofMillis(100));
        // refCount() 返回 Flux
        // refCount() 是 publish().refCount() 的简写形式。
        // 它将源 Flux 转换为 ConnectableFlux 并立即调用 connect()。
        // 它还将 refCount() 的订阅者数量设置为 n。
        // 这意味着只要有 n 个订阅者，就会保持连接。
        // 如果没有订阅者，它将取消连接。
        // 请注意，如果您在订阅者注册之前调用 connect()，则可能会丢失一些元素。
        Flux<Integer> refCount = flux.publish().refCount(2);
        refCount.subscribe(i -> log.info("subscribe1: {}", i));
        refCount.subscribe(i -> log.info("subscribe2: {}", i));
        TimeUnit.SECONDS.sleep(1);
    }

    /**
     * 但在取消之前给新订阅者一个机会：.publish（）.refCount（n， Duration）
     */
    @Test
    public void refCountDuration() throws InterruptedException {
        log.info("但在取消之前给新订阅者一个机会：.publish（）.refCount（n， Duration）");
        // 但在取消之前给新订阅者一个机会：.publish（）.refCount（n， Duration）
        Flux<Integer> flux = Flux.range(1, 100).delayElements(Duration.ofMillis(100));
        // refCount() 返回 Flux
        // refCount() 是 publish().refCount() 的简写形式。
        // 它将源 Flux 转换为 ConnectableFlux 并立即调用 connect()。
        // 它还将 refCount() 的订阅者数量设置为 n。
        // 这意味着只要有 n 个订阅者，就会保持连接。
        // 如果没有订阅者，它将取消连接。
        // 请注意，如果您在订阅者注册之前调用 connect()，则可能会丢失一些元素。
        // refCount(2, Duration.ofMillis(500)) 表示：只要有2个订阅者，就会保持连接，如果没有订阅者，它将在500毫秒后取消连接。
        Flux<Integer> refCount = flux.publish().refCount(2, Duration.ofMillis(500));
        refCount.subscribe(i -> log.info("subscribe1: {}", i));
        TimeUnit.SECONDS.sleep(1);
        refCount.subscribe(i -> log.info("subscribe2: {}", i));
        TimeUnit.SECONDS.sleep(1);
    }

    /**
     * 最多元素：cache（int）n
     */
    @Test
    public void cache() throws InterruptedException {
        log.info("最多元素：cache（int）n");
        // 最多元素：cache（int）n
        Flux<Integer> flux = Flux.range(1, 100).delayElements(Duration.ofMillis(100));
        // cache() 返回 Flux
        // cache() 是 cache(Integer.MAX_VALUE) 的简写形式。
        // 它将源 Flux 转换为 ConnectableFlux 并立即调用 connect()。
        // 它还将 refCount() 的订阅者数量设置为 1。
        // 这意味着只要有一个订阅者，就会保持连接。
        // 如果没有订阅者，它将取消连接。
        // 请注意，如果您在订阅者注册之前调用 connect()，则可能会丢失一些元素。
        // cache(2) 表示：最多缓存2个元素。
        Flux<Integer> cache = flux.cache(2);
        cache.subscribe(i -> log.info("subscribe1: {}", i));
        TimeUnit.SECONDS.sleep(1);
        cache.subscribe(i -> log.info("subscribe2: {}", i));
        TimeUnit.SECONDS.sleep(1);
    }

    /**
     * 缓存在持续时间（生存时间）内看到的最新元素：（Flux|Mono#cache(Duration))
     */
    @Test
    public void cacheDuration() throws InterruptedException {
        log.info("缓存在持续时间（生存时间）内看到的最新元素：（Flux|Mono#cache(Duration))");
        // 缓存在持续时间（生存时间）内看到的最新元素：（Flux|Mono#cache(Duration))
        Flux<Integer> flux = Flux.range(1, 100).delayElements(Duration.ofMillis(100));
        // cache() 返回 Flux
        // cache() 是 cache(Integer.MAX_VALUE) 的简写形式。
        // 它将源 Flux 转换为 ConnectableFlux 并立即调用 connect()。
        // 它还将 refCount() 的订阅者数量设置为 1。
        // 这意味着只要有一个订阅者，就会保持连接。
        // 如果没有订阅者，它将取消连接。
        // 请注意，如果您在订阅者注册之前调用 connect()，则可能会丢失一些元素。
        // cache(Duration.ofMillis(500)) 表示：缓存500毫秒内看到的最新元素。
        Flux<Integer> cache = flux.cache(Duration.ofMillis(500));
        cache.subscribe(i -> log.info("subscribe1: {}", i));
        TimeUnit.SECONDS.sleep(1);
        cache.subscribe(i -> log.info("subscribe2: {}", i));
        TimeUnit.SECONDS.sleep(1);
    }

    /**
     * 但保留不超过元素：cache（int， Duration）n
     */
    @Test
    public void cacheHistoryDuration() throws InterruptedException {
        log.info("但保留不超过元素：cache（int， Duration）n");
        // 但保留不超过元素：cache（int， Duration）n
        Flux<Integer> flux = Flux.range(1, 100).delayElements(Duration.ofMillis(100));
        // cache() 返回 Flux
        // cache() 是 cache(Integer.MAX_VALUE) 的简写形式。
        // 它将源 Flux 转换为 ConnectableFlux 并立即调用 connect()。
        // 它还将 refCount() 的订阅者数量设置为 1。
        // 这意味着只要有一个订阅者，就会保持连接。
        // 如果没有订阅者，它将取消连接。
        // 请注意，如果您在订阅者注册之前调用 connect()，则可能会丢失一些元素。
        // cache(2, Duration.ofMillis(500)) 表示：最多缓存2个元素，缓存500毫秒内看到的最新元素。
        Flux<Integer> cache = flux.cache(2, Duration.ofMillis(500));
        cache.subscribe(i -> log.info("subscribe1: {}", i));
        TimeUnit.SECONDS.sleep(1);
        cache.subscribe(i -> log.info("subscribe2: {}", i));
        TimeUnit.SECONDS.sleep(1);
    }

    /**
     * 但不会立即触发源：Flux#replay（返回 ConnectableFlux)
     */
    @Test
    public void replay() throws InterruptedException {
        log.info("但不会立即触发源：Flux#replay（返回 ConnectableFlux)");
        // 但不会立即触发源：Flux#replay（返回 ConnectableFlux)
        Flux<Integer> flux = Flux.range(1, 100).delayElements(Duration.ofMillis(100));
        // replay() 返回 ConnectableFlux
        // replay() 是 replay(Integer.MAX_VALUE) 的简写形式。
        // 它将源 Flux 转换为 ConnectableFlux 并立即调用 connect()。
        // 它还将 refCount() 的订阅者数量设置为 1。
        // 这意味着只要有一个订阅者，就会保持连接。
        // 如果没有订阅者，它将取消连接。
        // 请注意，如果您在订阅者注册之前调用 connect()，则可能会丢失一些元素。
        // replay() 表示：缓存所有元素。
        ConnectableFlux<Integer> replay = flux.replay();
        replay.subscribe(i -> log.info("subscribe1: {}", i));
        replay.subscribe(i -> log.info("subscribe2: {}", i));
        replay.connect();
        TimeUnit.SECONDS.sleep(1);
    }

}
